#ifndef _KAFKA_EVENTCB_H_H_H
#define _KAFKA_EVENTCB_H_H_H

#include <librdkafka/rdkafkacpp.h>

class RdKafkaEventCb : public RdKafka::EventCb
{
public:
    void event_cb(RdKafka::Event &event)
    {
        if (event.type() == RdKafka::Event::EVENT_ERROR) {
            if (event.fatal()) {
                fprintf(stderr, "rdkafka event fatal error\n");
            }
            fprintf(stderr, "rdkafka event error: %s\n", RdKafka::err2str(event.err()).c_str());
        } else if (event.type() == RdKafka::Event::EVENT_STATS) {
            fprintf(stdout, "rdkafka event stats: %s\n", event.str().c_str());
        } else if (event.type() == RdKafka::Event::EVENT_LOG) {
            fprintf(stderr, "rdkafka event log: %i-%s: %s\n", event.severity(), event.fac().c_str(), event.str().c_str());
        } else {
            fprintf(stderr, "rdkafka event %d: %s\n", event.type(), RdKafka::err2str(event.err()).c_str());
        }
    }
};

#endif // _KAFKA_EVENTCB_H_H_H